package com.cxb.demotest.mq.rocketmq.part_by_order;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 *顺序消费
 *     指定消息发送到同一个topic中，并且topic中只有一个queue，即可保证顺序消费
 *部分顺序消费
 *     根据每个订单消息通过一定的特征放在特定的队列中，保证消息的顺序性
 * @author Xb
 */
public class ConsumerInOrder {
    public static void main(String[] args) throws MQClientException {
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderConsumer");

        // 指定namesrv
        consumer.setNamesrvAddr("127.0.0.1:9876");

        // 消费者从哪个位置消费，此处为第一次启动从队列最后位置消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        // 订阅topic
        consumer.subscribe("PartOrder", "*");

        // 注册回调函数、处理消息
        consumer.registerMessageListener(new MessageListenerOrderly() {
            Random random = new Random();
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
                for (MessageExt msg : msgs) {
                    // 可以看到每个queue有唯一的consume线程消费，订单对每个queue分区有序
                    System.out.println("consumeThread=" + Thread.currentThread().getName()
                    + ",queueId=" + msg.getQueueId() + ",context=" + new String(msg.getBody()));
                }
                try {
                    // 模拟业务逻辑处理
                    TimeUnit.MILLISECONDS.sleep(random.nextInt(300));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    // 此处需要注意！！！！！！要先等一会，再处理这些消息，而不是放到重试队列里，否则会破坏消息消费的顺序性
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();

        // 注销消费者
        System.out.println("Consume completed.");


    }
}
